//package org.ddd.pattern.event.springboot.rocketmq.publish.tt;
//
//
//import lombok.SneakyThrows;
//import org.ddd.pattern.event.domain.DomainEventRepository;
//import org.ddd.pattern.event.domain.DomainEventTrackerRepository;
//import org.ddd.pattern.event.domain.data.DomainEventTracker;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
//
//import java.text.MessageFormat;
//import java.util.List;
//import java.util.Objects;
//import java.util.concurrent.CompletableFuture;
//import java.util.concurrent.ConcurrentHashMap;
//import java.util.concurrent.ThreadPoolExecutor;
//import java.util.concurrent.TimeUnit;
//import java.util.concurrent.locks.ReentrantLock;
//
//
////public class RocketMQEventPublisher implements IEventPublisher {
//
////    Logger log = LoggerFactory.getLogger(RocketMQEventPublisher.class);
////    private final DomainEventRepository eventQueryStore;
////    private final DomainEventTrackerRepository trackerStore;
////    private final IEventDispatcher dispatcher;
////    private final ThreadPoolTaskExecutor executor;
////    private final ConcurrentHashMap<String, Boolean> trackerTaskStatus = new ConcurrentHashMap<>(100);
////    private final ConcurrentHashMap<String, ReentrantLock> trackerTaskLocks = new ConcurrentHashMap<>(100);
////
////    public RocketMQEventPublisher(
////            DomainEventRepository eventQueryStore,
////            DomainEventTrackerRepository trackerStore,
////            IEventDispatcher dispatcher) {
////        this.executor = newThreadPoolTaskExecutor();
////        this.eventQueryStore = Objects.requireNonNull(eventQueryStore);
////        this.trackerStore = Objects.requireNonNull(trackerStore);
////        this.dispatcher = Objects.requireNonNull(dispatcher);
////    }
//
////    private synchronized ThreadPoolTaskExecutor newThreadPoolTaskExecutor() {
////        if (this.executor != null) {
////            return this.executor;
////        }
////        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
////        taskExecutor.setCorePoolSize(15);//核心线程数
////        taskExecutor.setMaxPoolSize(120);//线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程数的线程
////        taskExecutor.setQueueCapacity(3000);//缓存队列
////        taskExecutor.setKeepAliveSeconds(200);//许的空闲时间,当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
////        taskExecutor.setThreadNamePrefix("PublishableEventTaskExecutor");//异步方法内部线程名称
////        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
////        taskExecutor.initialize();
////        return taskExecutor;
////    }
//
////    @Override
////    public void publish(PublishableEventContext context) {
////        doPublish(context);
////    }
//
////    @SneakyThrows
////    protected void doPublish(PublishableEventContext context) {
////        long now = System.currentTimeMillis();
////        List<DomainEventTracker> publishedNotificationTrackers = this.trackerStore.getTrackers();
////
////        CompletableFuture<Boolean> rootFuture = CompletableFuture.completedFuture(false);
////
////        //如果存在就不发
////        for (DomainEventTracker tracker : publishedNotificationTrackers) {
////            //只有值为 true 的情况下才退出
////            if (this.trackerTaskStatus.getOrDefault(tracker.getDomainType(), false)) {
////                continue;
////            }
////
////            CompletableFuture<Boolean> completableFuture = CompletableFuture.supplyAsync(() -> {
////                ReentrantLock reentrantLock = this.trackerTaskLocks.computeIfAbsent(tracker.getDomainType(), s -> new ReentrantLock());
////
////                reentrantLock.lock();
////                try {
////                    publishGroupByTag(tracker, context);
////                    return false;
////                } catch (Exception ex) {
////                    log.error("【领域事件】发送消息异常,消费方注意幂等处理", ex);
////                } finally {
////                    //只有 值为 false 的情况下才删除
////                    this.trackerTaskStatus.remove(tracker.getDomainType(), false);
////                    reentrantLock.unlock();
////                }
////                return true;
////            }, executor).exceptionally(val -> true);
////
////            rootFuture = rootFuture.thenCompose(val1 -> completableFuture.thenApply(val2 -> val1 || val2).exceptionally(val -> true));
////        }
////
////        if (rootFuture.get()) {
////            throw new RuntimeException("消息发送存在异常..");
////        }
////
////        long costTime = System.currentTimeMillis() - now;
////        log.trace("发送消息花费时间: {} ms", costTime);
////    }
//
////    @Override
////    public void publish(String tagName, PublishableEventContext context) {
////        long now = System.currentTimeMillis();
////        DomainEventTracker domainEventTracker = this.trackerStore.getTrackerByTagName(tagName);
////        try {
////            publishGroupByTag(domainEventTracker, context);
////        } catch (Exception ex) {
////            log.error("消息发送失败", ex);
////            throw new RuntimeException();
////        }
////        long costTime = System.currentTimeMillis() - now;
////        log.trace("发送消息花费时间: {} ms", costTime);
////    }
//
////    private int publish(PublishedEventTracker tracker, List<StoredEvent> storedEvents) {
////        return publishWithOrderly(storedEvents);
////    }
//
////    protected void publishGroupByTag(DomainEventTracker tracker, PublishableEventContext context) {
////        int batchSize = 50;
////        List<StoredEvent> storedEvents = this.eventQueryStore.findByTagName(tracker.getDomainType(), tracker.getLatestStoredId(), batchSize);
////        int totalPublishCount = 0;
////
////        if (storedEvents.isEmpty()) {
////            this.trackerTaskStatus.putIfAbsent(tracker.getDomainType(), false); //只有没有值的情况才写 false
////        } else {
////            this.trackerTaskStatus.put(tracker.getDomainType(), true); // true 值，一定写入
////        }
////
////        while (!storedEvents.isEmpty()) {
////            //      if (log.isDebugEnabled()) {
//////        for (StoredEvent storedEvent : storedEvents) {
//////          log.debug("准备发送事件,TOPIC:{},事件编码：{},From-TrackerId:{}", storedEvent.getTopic(),
//////              storedEvent.getEventId(), tracker.getMostRecentStoredId());
//////        }
//////      }
//////      storedEvents.stream().map(x->x.getEventId()).collect(Collectors.toList());
////            int size = storedEvents.size();
////            log.debug("【领域事件】消息发送成功, 领域[{}]本次预计发送事件数量:{}", tracker.getDomainType(), size);
////
////            //发送到rocketmq 中
////            int publishCount = publishWithOrderly(storedEvents);
////            totalPublishCount += publishCount;
////
////            if (publishCount > 0) {
////                StoredEvent lastStoredEvent = storedEvents.get(publishCount - 1);
////                DomainEventTracker newPublishedNotificationTracker = new DomainEventTracker();
////                newPublishedNotificationTracker.setDomainType(tracker.getDomainType());
////                newPublishedNotificationTracker.setLatestStoredId(lastStoredEvent.getId());
////                //todo 死机有可能会重复发送，最多 takeCount 条，即 50 条
////                boolean result = this.trackerStore.updateTrack(newPublishedNotificationTracker);
////                if (!result) {
////                    //todo 获取最近的 storeId
////                    log.warn("trackerStore.updateTrack 返回值为 false ");
////                }
////                storedEvents = this.eventQueryStore.findByTagName(tracker.getDomainType(), lastStoredEvent.getId(), batchSize);
////                this.trackerTaskStatus.put(tracker.getDomainType(), !storedEvents.isEmpty());
////                tracker = newPublishedNotificationTracker;
////                log.debug("【领域事件】消息发送成功, 领域[{}]本次发送事件数量:{}, 期望发送事件数量:{}, 保存最新TrackerId:{}", tracker.getDomainType(), publishCount, size, lastStoredEvent.getId());
////            } else {
////                log.debug("【领域事件】未发送任何消息, 领域[{}]尝试重新发送事件", tracker.getDomainType());
////                try {
////                    //发送失败后3秒后重试 , todo  随机数 1-3 秒
////                    TimeUnit.SECONDS.sleep(2);
////                    storedEvents = this.eventQueryStore.findByTagName(tracker.getDomainType(), tracker.getLatestStoredId(), batchSize);
////                } catch (InterruptedException e) {
////                    log.warn("消息发送失败", e);
////                }
////            }
////        }
////
////        if (totalPublishCount > 0) {
////            log.debug("【领域事件】消息发送成功, 领域[{}]本次发送事件总量:{}", tracker.getDomainType(), totalPublishCount);
////        }
////
////    }
//
////    private int publishWithOrderly(List<StoredEvent> storedEvents) {
////        int i = 0;
////        try {
////            for (StoredEvent storedEvent : storedEvents) {
////                //TODO 批量发送？？ 何挺强
////                final EventDispatchResult dispatchResult = this.dispatcher.dispatch(storedEvent);
////                if (dispatchResult.getDispatchStatus() == DispatchStatus.FALIURE) {
////                    return i;
////                    //throw new DispatchingEventException("发送消息失败");
////                }
////                //TODO 异步更新，数据库状态 为4 ，并发送时间 何挺强
////                i++;
////            }
////        } catch (Throwable ex) {
////            log.error(
////                    MessageFormat.format("消息发送失败, 错误消息 -> {0}, 消息编码 -> {1}", ex.getMessage(),
////                            storedEvents.get(i).getEventKey()), ex);
////            return i;
////        }
////        return i;
////    }
//
////}
